package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class test7 {
    public static void main(String[] args) throws Exception {
        //流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //检查点
        env.enableCheckpointing(5000);
        //kafka读取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test7");
        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));

        SingleOutputStreamOperator<String> orderInfo = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));
        SingleOutputStreamOperator<String> orderCargo = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));

        //1.将order_cargo的数据与order_info数据流通过订单id进行关联。（由于存在数据乱序或者延时的情况，导致两个流的数据进度不一致，
        // 请酌情使用合适的join方式，二者数据前后相差最大5秒，同时order_info状态的ttl设置为5s）（10分）
        //orderInfo.print();
        //orderCargo.print();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SingleOutputStreamOperator<Order_info> orderInfoDS = orderInfo.map(new MapFunction<String, Order_info>() {
            @Override
            public Order_info map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject data = jsonObject.getJSONObject("data");
                return JSON.parseObject(data.toJSONString(), Order_info.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Order_info>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                //create_time=2022-07-12T02:54:36Z
                .withTimestampAssigner((event, timestamp) -> {
//                    String s = event.getCreate_time().replace("T", " ");
//                    //System.out.println(s);
//                    String ss = s.replace("Z", "");
                    //System.out.println(ss);
                    try {
                        return sdf.parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
        //orderInfoDS.print();

        SingleOutputStreamOperator<Order_cargo> orderCagoDS = orderCargo.map(new MapFunction<String, Order_cargo>() {
            @Override
            public Order_cargo map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject data = jsonObject.getJSONObject("data");
                return JSON.parseObject(data.toJSONString(), Order_cargo.class);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Order_cargo>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                //create_time=2022-07-12T02:54:36Z
                .withTimestampAssigner((event, timestamp) -> {
//                    String s = event.getCreate_time().replace("T", " ");
//                    //System.out.println(s);
//                    String ss = s.replace("Z", "");
                    //System.out.println(ss);
                    try {
                        return sdf.parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));

        SingleOutputStreamOperator<Tuple2<Order_info, Order_cargo>> unionDS = orderInfoDS
                .keyBy(x -> x.getId())
                .intervalJoin(orderCagoDS.keyBy(x -> x.getOrder_id()))
                .between(Time.seconds(-5), Time.seconds(0))
                .process(new ProcessJoinFunction<Order_info, Order_cargo, Tuple2<Order_info, Order_cargo>>() {
                    @Override
                    public void processElement(Order_info left, Order_cargo right, ProcessJoinFunction<Order_info, Order_cargo, Tuple2<Order_info, Order_cargo>>.Context ctx, Collector<Tuple2<Order_info, Order_cargo>> out) throws Exception {
                        out.collect(new Tuple2<>(left, right));
                    }
                });
        //unionDS.print();
        //(Order_info(id=10061, order_no=2024-1100355484-00000140, status=60020, collect_type=69002,
        // user_id=142, receiver_complex_id=86, receiver_province_id=310000, receiver_city_id=310100,
        // receiver_district_id=310114, receiver_address=上海市市辖区嘉定区湖夏城3号楼123, receiver_name=宇文凡佳,
        // receiver_phone=13422919718, receive_location=32.123123,71.12341, sender_complex_id=80,
        // sender_province_id=520000, sender_city_id=520100, sender_district_id=520102,
        // sender_address=贵州省贵阳市南明区森凯苑3号楼435, sender_name=乐瑞, sender_phone=13267752128,
        // send_location=null, payment_type=62002, cargo_num=1, amount=13.00, estimate_arrive_time=1658181276000,
        // distance=2090.00, create_time=2022-07-12T02:54:36Z, update_time=2022-07-12T02:56:37Z, is_deleted=0),
        // Order_cargo(id=10059, order_id=10061, cargo_type=74002, volume_length=null, volume_width=null,
        // volume_height=null, weight=4.00,
        // remark=null, create_time=2022-07-12T02:54:36Z, update_time=null, is_deleted=0))






        env.execute();
    }
}
